/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */
package com.sys.midware.hbase.hcore.Handler;

import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;

import com.sys.midware.assist.utils.string.StringUtil;
import com.sys.midware.hbase.config.BHBaseConfig;
import com.sys.midware.hbase.config.BHBaseConfigHolder;
import com.sys.midware.hbase.config.BHBaseConfigMapHolder;
import com.sys.midware.hbase.config.BHBaseConfigUtil;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 
 * <b>ClassName：</b> HBaseDBFactory <br/>
 * <b>Description：</b> TODO <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月26日 下午3:40:39 <br/>
 * <b>@version: </b> <br/>
 */
public class HBaseDBFactory {
    private static final Logger logger = LoggerFactory.getLogger(HBaseDBFactory.class);

    private static HBaseDBFactory facotry = new HBaseDBFactory();

    private HBaseDBFactory() {

    }

    public static HBaseDBFactory createFactory() {

        return facotry;
    }

    private Configuration initHBaseConfiguration(String nodeName) {
        BHBaseConfigUtil.initConfig();

        Map<String, BHBaseConfig> configMap = BHBaseConfigMapHolder.get();
        BHBaseConfig config = null;
        if (StringUtil.isNotEmpty(nodeName)) {
            config = configMap.get(nodeName);
            BHBaseConfigHolder.put(config);
        } else {
            // 如果没有传nodename，则随机取一个hbase配置
            for (Entry<String, BHBaseConfig> map : configMap.entrySet()) {
                config = map.getValue();
                BHBaseConfigHolder.put(config);
                if (map.getValue() == null) {
                    continue;
                }
            }
        }
        // 初始化hadoopHBase配置
        Configuration HBASE_CONFIG = HBaseConfiguration.create();
//        HBASE_CONFIG.set("hbase.master", config.getHbaseHost());
//        HBASE_CONFIG.set("hbase.master.port", String.valueOf(config.getHbasePort()));
//        HBASE_CONFIG.set("hbase.zookeeper.quorum", config.getZKCluster());
//        HBASE_CONFIG.set("hbase.zookeeper.property.clientPort", String.valueOf(config.getZKPort()));
//        HBASE_CONFIG.set("fs.defaultFS","hdfs://ns");
//        HBASE_CONFIG.set("zookeeper.session.timeout", "180000");
        
        HBASE_CONFIG.set("yarn.resourcemanager.hostname",config.getHbaseHost());
        HBASE_CONFIG.set("hbase.master.port", String.valueOf(config.getHbasePort()));
        HBASE_CONFIG.set("hbase.zookeeper.quorum",config.getZKCluster());
        HBASE_CONFIG.set("hbase.zookeeper.property.clientPort", String.valueOf(config.getZKPort()));
        return HBASE_CONFIG;
    }

    public DBHandler getDBHandler(String tableName) throws Exception {
        Configuration HBASE_CONFIG = initHBaseConfiguration(null);
        if (tableName instanceof String) {
            return new HBaseDBHandler((String) tableName, HBASE_CONFIG);
        } else
            throw new IllegalArgumentException("arugument key is not String :" + tableName.getClass().getName());
    }

    public DBHandler getDBHandler(String nodeName, String tableName) throws Exception {
        Configuration HBASE_CONFIG = initHBaseConfiguration(nodeName);
        if (tableName instanceof String) {
            return new HBaseDBHandler((String) tableName, HBASE_CONFIG);
        } else
            throw new IllegalArgumentException("arugument key is not String :" + tableName.getClass().getName());
    }

    public void shutdown() throws Exception {
        facotry = null;
    }

    /**
     * 校验HBase是否可用
     * 
     * @param
     * @return void
     */
    public void checkDBAvailable(String nodeName) throws Exception {
        try {
            Configuration HBASE_CONFIG = initHBaseConfiguration(nodeName);
            HBaseAdmin.checkHBaseAvailable(HBASE_CONFIG);
        } catch (MasterNotRunningException e) {
            logger.error(e.getMessage(), e);
            throw e;
        } catch (ZooKeeperConnectionException e) {
            logger.error(e.getMessage(), e);
            throw e;
        }
    }

}
